feat(kinesis): Kinesis Data Streams Library by ekjotmultani · Pull Request #6516 · aws-amplify/amplify-flutter

ekjotmultani · 2026-01-22T17:21:45Z

Issue #, if available:

Description of changes:

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/record_storage.dart

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/kinesis_sender.dart

packages/kinesis/aws_kinesis_datastreams/lib/src/exception/kinesis_exception.dart

jvh-aws · 2026-02-05T10:12:21Z

packages/kinesis/aws_kinesis_datastreams_dart/lib/src/impl/auto_flush_scheduler.dart

+  final Future<void> Function() _onFlush;
+  Timer? _timer;
+  bool _enabled = true;
+  bool _closed = false;


What is the benefit of having both _enabled and _closed?

_closed can serve as a flag for other code to know whether or not the client itself is shut down or if just flushing is enabled or not. In our implementation they do similar things but it is useful for getters to know the internal state of the client I feel. What do you think?

I think both approaches are fine, but personally, I recommend merging them for the auto flush scheduler, since _closed is not really used.

packages/kinesis/aws_kinesis_datastreams_dart/lib/src/impl/kinesis_sender.dart

packages/kinesis/aws_kinesis_datastreams/test/e2e/kinesis_e2e_test.dart

packages/kinesis/aws_kinesis_datastreams_dart/test/e2e/kinesis_e2e_test.dart

packages/kinesis/aws_kinesis_datastreams/test/e2e/kinesis_e2e_test.dart

jvh-aws · 2026-02-05T10:34:36Z

packages/kinesis/aws_kinesis_datastreams_dart/test/record_storage_test.dart

+      });
+    });
+  });
+}


Can we ensure concurrent access to the record storage works via a test?

packages/amplify_core/lib/src/category/amplify_categories.dart

packages/kinesis/aws_kinesis_datastreams_dart/lib/src/db/kinesis_record_database.dart

jvh-aws · 2026-03-04T09:45:57Z

packages/kinesis/aws_kinesis_datastreams/lib/src/exception/amplify_kinesis_exception.dart

+/// {@template aws_kinesis_datastreams.kinesis_service_exception}
+/// Thrown when a Kinesis SDK/API error occurs. Inspect [sdkException] for details.
+/// {@endtemplate}
+final class KinesisServiceException extends AmplifyKinesisException {


I think we discussed that those errors are silently swallowed and not exposed.

jvh-aws · 2026-03-04T09:48:03Z

packages/kinesis/aws_kinesis_datastreams/lib/src/exception/amplify_kinesis_exception.dart

+/// Thrown when a single record exceeds the Kinesis per-record size limit
+/// (10 MiB, partition key + data blob combined).
+/// {@endtemplate}
+final class KinesisRecordTooLargeException extends AmplifyKinesisException {


If we have a specific type for records being too large, we should also have one for the partition key being too long.

agreed, added a corresponding type

jvh-aws · 2026-03-04T09:54:59Z

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/record_client.dart

+    if (_closed) throw ClientClosedException();
+    if (!_enabled) return;
+
+    if (record.dataSize > kKinesisMaxRecordBytes) {


We should also check the partition key length here. It should be in (0, 256]

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/record_client.dart

jvh-aws · 2026-03-04T10:02:58Z

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/record_storage.dart

+      ''',
+      variables: [Variable.withInt(maxCount), Variable.withInt(maxBytes)],
+      readsFrom: {_db.kinesisRecords},
+    ).get();


What happens if this query fails? In my understanding, currently, we'd swallow the error in flush's try-catch block. However, I think we would like to expose such errors as storage errors. Can we add logic to expose such errors as Kinesis* errors? Or am I missing something?

no, good call, wrapped the failures in error object

jvh-aws · 2026-03-04T10:07:38Z

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/record_storage.dart

+
+  /// Returns the current total size of cached data in bytes.
+  Future<int> getCurrentCacheSize() async {
+    final query = _db.selectOnly(_db.kinesisRecords)


I think, I commented this before: We can keep an integer for the cache size and update / check against that, when we add new records. We will only need to run a DB query after removing records. This allows us to avoid the extra query when recording, which is most likely the most frequent operation.

Adds the foundational package with minimal constructs needed for the Kinesis client libraries: - AmplifyException base class - AWSCredentialsProvider<T> and AWSCredentials sealed hierarchy - Logger interface with AmplifyLogger implementation - LogLevel enum - Result<T, E> sealed type

Provides V2CredentialsProviderBridge to bridge aws_common (V2) credentials to amplify_foundation_dart (V3) credentials. Shared by kinesis client packages to avoid duplicating bridge logic.

* feat(kinesis): return RecordData from record() for cross-platform alignment\n\nChange record() return type from Result<void> to Result<RecordData>\nacross both the Dart client and Flutter wrapper. RecordData contains\nthe recorded entry size (data + partition key bytes), aligning with\nthe Android and Swift Kinesis client APIs.\n\nWhen the client is disabled, returns RecordData(recordSize: 0)." * feat(kinesis): return RecordData from record() for cross-platform alignment\n\nChange record() return type from Result<void> to Result<RecordData>\nacross both the Dart client and Flutter wrapper. RecordData contains\nthe recorded entry size (data + partition key bytes), aligning with\nthe Android and Swift Kinesis client APIs.\n\nWhen the client is disabled, returns RecordData(recordSize: 0)."

refactor(kinesis): rename kinesis_data_streams_options.dart to amplify_kinesis_client_options.dart Align the options file name with the client class name for consistency. Co-authored-by: jvh-aws <jvhoff@amazon.de>

* Reapply "Align exceptions with Amplify v2 AmplifyException pattern\n\nSwitch AmplifyKinesisException to extend amplify_core's\nAmplifyException instead of amplify_foundation_dart's.\nAdopt positional message, optional recoverySuggestion,\nunderlyingException (was cause), const constructors,\nand runtimeTypeName overrides."" This reverts commit f660c5f. * Add const to ClientClosedException() call sites --------- Co-authored-by: jvh-aws <jvhoff@amazon.de>

Expose library version in foundation, use it in kinesis user agent

* Expose library version in foundation, use it in kinesis user agent * Fix flush strategy naming * Apply autoformat * Apply autoformat

* Expose library version in foundation, use it in kinesis user agent * Fix flush strategy naming * Apply autoformat * Apply autoformat * Rename package from aws_kinesis_datastreams to amplify_kinesis * Fix analyzer

This reverts commit c3cfb6c.

Strip RecordData to an empty wrapper for now. Fields can be added later once cross-platform alignment on the return shape is finalized.

* Minor doc fixes and remove reexport * Adapt public facing docstrings

…6795) * refactor: remove Category.kinesis enum, use string categories in deploy script Decouple deploy_gen2.dart from the amplify_core Category enum by changing AmplifyBackendGroup.category to a plain String. This allows removing Category.kinesis from the enum and its dead switch case in AmplifyClassImpl, since the kinesis client is standalone and does not participate in the Amplify plugin registration system. The --category CLI flag now derives allowed values from infraConfig instead of Category.values. * fix: match original Category.api.name casing ('API' not 'Api')

ekjotmultani requested a review from a team as a code owner January 22, 2026 17:21

ekjotmultani marked this pull request as draft January 26, 2026 19:13

ekjotmultani changed the title ~~Feat/kinesis data streams library~~ feat(kinesis): Kinesis Data Streams Library Feb 3, 2026

jvh-aws reviewed Feb 5, 2026

View reviewed changes

ekjotmultani force-pushed the feat/kinesis-data-streams-library branch 3 times, most recently from a21d70d to ff90989 Compare February 27, 2026 08:29

ekjotmultani marked this pull request as ready for review February 27, 2026 08:29

ekjotmultani force-pushed the feat/kinesis-data-streams-library branch 4 times, most recently from d9ec047 to 8ec0ec5 Compare March 4, 2026 06:14

jvh-aws reviewed Mar 4, 2026

View reviewed changes

packages/amplify_core/lib/src/category/amplify_categories.dart Outdated Show resolved Hide resolved

jvh-aws reviewed Mar 4, 2026

View reviewed changes

packages/kinesis/aws_kinesis_datastreams_dart/lib/src/db/kinesis_record_database.dart Outdated Show resolved Hide resolved

jvh-aws reviewed Mar 4, 2026

View reviewed changes

packages/kinesis/aws_kinesis_datastreams/lib/src/impl/record_client.dart Outdated Show resolved Hide resolved

jvh-aws reviewed Mar 4, 2026

View reviewed changes

ekjotmultani changed the base branch from main to feat/amplify-foundation-dart-minimal March 4, 2026 17:08

ekjotmultani force-pushed the feat/kinesis-data-streams-library branch from 5ffac53 to 11bb6e0 Compare March 5, 2026 15:04

ekjotmultani changed the base branch from feat/amplify-foundation-dart-minimal to main March 5, 2026 16:47

ekjotmultani and others added 8 commits March 5, 2026 08:49

feat: add amplify_foundation_dart_bridge package

9a5dcf6

Provides V2CredentialsProviderBridge to bridge aws_common (V2) credentials to amplify_foundation_dart (V3) credentials. Shared by kinesis client packages to avoid duplicating bridge logic.

Fix LICENSE files

4d49b63

Review fixes

30648b1

Fix dependencies

a0cc227

Satisfy the linter

9c94ed2

Move packages

0d927e6

Refactor logging

f47ba9f

jvh-aws and others added 19 commits March 13, 2026 14:50

Add licenses

a6650c5

Apply autoformat

fea3e68

Revert "Add approved email sender"

03171e0

Add pre sign up handler

14ee877

Delete user fater tests

56cd6c6

Remove log sink from e2e test

cdddb7b

Fix/kinesis rename options file (#6789)

1e9a2a4

refactor(kinesis): rename kinesis_data_streams_options.dart to amplify_kinesis_client_options.dart Align the options file name with the client class name for consistency. Co-authored-by: jvh-aws <jvhoff@amazon.de>

Kinesis: Fix user agents (#6790)

533b626

Expose library version in foundation, use it in kinesis user agent

Jv/kinesis fix flush strategy naming (#6791)

380a26f

* Expose library version in foundation, use it in kinesis user agent * Fix flush strategy naming * Apply autoformat * Apply autoformat

Jv/kinesis rename package to amplify kinesis (#6792)

203eb02

* Expose library version in foundation, use it in kinesis user agent * Fix flush strategy naming * Apply autoformat * Apply autoformat * Rename package from aws_kinesis_datastreams to amplify_kinesis * Fix analyzer

Fix dependabot.yaml

619a449

Fix dependabot.yaml (2)

3238d68

Fix dependabot.yaml (3)

c3cfb6c

Revert "Fix dependabot.yaml (3)"

3b8a29f

This reverts commit c3cfb6c.

Regenerate workflows

ca786fa

Fix dependabot config

d54d98e

Fix dependabot config

bfb3f2b

harsh62 previously approved these changes Mar 16, 2026

View reviewed changes

harsh62 requested a review from cadivus March 16, 2026 17:34

Jonas Greifenhain and others added 2 commits March 16, 2026 18:53

Fix renaming package

5fa9abb

refactor(kinesis): remove recordSize from RecordData

b50ea0c

Strip RecordData to an empty wrapper for now. Fields can be added later once cross-platform alignment on the return shape is finalized.

ekjotmultani dismissed a stale review via b50ea0c March 16, 2026 18:26

jvh-aws and others added 2 commits March 16, 2026 23:07

Minor doc fixes and remove reexport (#6794)

dfdf7a5

* Minor doc fixes and remove reexport * Adapt public facing docstrings

jvh-aws requested review from cadivus, harsh62 and jvh-aws March 17, 2026 09:27

jvh-aws merged commit 7f52a04 into main Mar 17, 2026
152 checks passed

Conversation

ekjotmultani commented Jan 22, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ekjotmultani Feb 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ekjotmultani Feb 13, 2026 •

edited

Loading